#include "../lib/transport.hpp"
#include "../lib/speaker.hpp"
#include "../lib/listner.hpp"

using namespace kport;
using pland::utill_to_string;
using pland::utill_log;
using pland::Exeption;

static void prodListner(int portno)
{
    int sockfd;
    struct sockaddr_in serv_addr;
    struct hostent *server;
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    server = gethostbyname("127.0.0.1");
    bzero((char *) &serv_addr, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    bcopy((char *)server->h_addr,
          (char *)&serv_addr.sin_addr.s_addr,
          server->h_length);
    serv_addr.sin_port = htons(portno);
    connect(sockfd,(struct sockaddr *) &serv_addr,sizeof(serv_addr));
    boost::this_thread::sleep(boost::posix_time::milliseconds(10));
    close(sockfd);
}


Transport::Transport(std::string config) :
    transportStatistics_(TRACE_BACK_LENGTH)
{
    running_ = false;
    listening_ = false;
    RoutingTable *rtable = RoutingTable::getInstance();
    rtable->loadTable(config);
    loadSettings_(config);
}

Transport::~Transport()
{
    stop();
    utill_log(std::string(__FILE__) + " Deleting Transport");
    for(size_t i =0; i< workItems_.size();i++)
    {
        if(workItems_[i].second->type() == WorkItem::IN)
            delete (Listner*)workItems_[i].second;
        else if(workItems_[i].second->type() == WorkItem::OUT)
            delete (Speaker*)workItems_[i].second;
        delete workItems_[i].first;
    }
}


void Transport::loadQueues(std::vector <pland::Chain::Chain_Queue > queues)
{
    for(size_t i = 0; i < queues.size(); i++)
    {
        queues_.insert(std::make_pair(queues[i].net, queues[i]));
    }
}

bool Transport::start()
{
    for(int i = 0; i< rules_.k; i++)
    {
        boost::thread *t1 = new boost::thread(&Transport::workerThread_, this);
        workers_.add_thread(t1);
    }
    running_ = true;
    return true;
}

bool Transport::stop()
{
    if(running_ == false)
    {
        return false;
    }
    workers_.interrupt_all();;
    workers_.join_all();

    running_ = false;
    return true;

}

void Transport::loadSettings_(std::string config_xml)
{
    TiXmlDocument doc(config_xml);
    if(!doc.LoadFile())
    {
        utill_log(std::string(__FILE__) + std::string("Bad config file ") + config_xml);
        throw Exeption("Bad config file " + config_xml, PEXP_BAD_CONFIG | PEXP_FATAL);
    }
    TiXmlNode *root = doc.RootElement();
    TiXmlNode *listener = root->FirstChild("listener");
    if(listener == NULL)
    {
        utill_log(std::string(__FILE__) + std::string("Bad config file ") + config_xml + " no listener section ");
        throw Exeption("Bad config file " + config_xml, PEXP_BAD_CONFIG | PEXP_FATAL);
    }
    if(listener->FirstChild("port") == NULL)
    {
        utill_log(std::string(__FILE__) + "routing table: " + config_xml +  " no port in listener section");
        throw Exeption("Bad config file " + config_xml, PEXP_BAD_CONFIG | PEXP_FATAL);
    }
    port_ = pland::utill_to_int(listener->FirstChild("port")->FirstChild()->Value());
    TiXmlNode *k = root->FirstChild("k");
    if(k == NULL)
    {
        utill_log(std::string(__FILE__) + std::string("Bad config file ") + config_xml + " no k section ");
        throw Exeption("Bad config file " + config_xml, PEXP_BAD_CONFIG | PEXP_FATAL);
    }
    rules_.k = pland::utill_to_int(k->FirstChild()->Value());
    TiXmlNode *bandwidth = root->FirstChild("bandwidth");
    if(bandwidth == NULL)
    {
        utill_log(std::string(__FILE__) + std::string("Bad config file ") + config_xml + " no bandwidth section section ");
        throw Exeption("Bad config file " + config_xml, PEXP_BAD_CONFIG | PEXP_FATAL);
    }
    rules_.bandwidth = pland::utill_to_int(bandwidth->FirstChild()->Value());
}

bool Transport::startListner()
{
    if(listening_ == true)
    {
        return false;
    }
    listening_ = true;
    listener_ = boost::thread(&Transport::listenerThread_, this);
    return listening_;
}

bool Transport::stopListner()
{
    if(listening_ == false)
    {
        return false;
    }
    listener_.interrupt();
    prodListner(port_);
    listener_.join();
    listening_ = false;
    return true;
}

void Transport::listenerThread_()
{
    int sockfd = -1;
    int newsockfd;

    try
    {
        socklen_t clien;
        struct sockaddr_in serv_addr, cli_addr;

        sockfd = ::socket(AF_INET, SOCK_STREAM, 0);

        bzero((char *) &serv_addr, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_addr.s_addr = INADDR_ANY;
        serv_addr.sin_port = htons(port_);

        int optval = 1;
        ::setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));

        if(::bind(sockfd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0)
        {
            utill_log(std::string(__FILE__) + " Transport: could not bind to port " + utill_to_string(port_) + ". Errno " + utill_to_string(errno));
            return;
        }

        if(::listen(sockfd,20) < 0)
        {
            utill_log(std::string(__FILE__) + " Transport: could not listen on port " + utill_to_string(port_) + ". Errno " + utill_to_string(errno));
            return;
        }

        utill_log(std::string(__FILE__) + " Transport: listening on port " + utill_to_string(port_));
        clien = sizeof(cli_addr);
        while((newsockfd = accept(sockfd,(struct sockaddr *) &cli_addr, &clien)))
        {
            boost::this_thread::interruption_point();
            char netName[NAME_MAX_LENGTH];
            if(newsockfd < 0)
            {
                utill_log(std::string(__FILE__) + " Transport: could not accept on port " + utill_to_string(port_) + ". Errno " + utill_to_string(errno));
                ::close(newsockfd);
                continue;
            }
            if(::read(newsockfd,netName, NAME_MAX_LENGTH) < 0)
            {
                utill_log(std::string(__FILE__) + " Transport: could not accept on port " + utill_to_string(port_) + ". Errno " + utill_to_string(errno));
                ::close(newsockfd);
                continue;
            }
            if(queues_.find(std::string(netName)) == queues_.end())
            {
                pland::utill_io_write(newsockfd,(char*)"NO",3);
                utill_log(std::string(__FILE__) + " Transport: Asked for net " + netName + " but it does not exist on this host");
                ::close(newsockfd);
                continue;
            }
            pland::utill_io_write(newsockfd,(char*)"OK",3);
            utill_log(std::string(__FILE__) + " Transport: Listner connected to net " + std::string(netName));
            Listner * w = new Listner(newsockfd,queues_[std::string(netName)].queue);
            boost::unique_lock< boost::shared_mutex > lock(workerManipulator_);
            workItems_.push_back(std::make_pair(new boost::mutex(), w));
        }
    }
    catch(boost::thread_interrupted e)
    {
        listening_ = false;
        if(sockfd > 0)
            close(sockfd);
        return;
    }
    return;
}

bool Transport::connectOutbound()
{
    RoutingTable *rtable = RoutingTable::getInstance();
    std::vector <std::string> routes = rtable->outGoingNets();
    bool success = true;
    for(size_t i =0; i < routes.size(); i++)
    {
        if(!connectRemote_(routes[i]))
            success = false;
    }
    return success;
}

bool Transport::connectRemote_(std::string netName)
{
    pland::Chain::Chain_Queue q = queues_[netName];
    char responce[3] = "NO";
    RoutingTable *rtable = RoutingTable::getInstance();
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    RoutingTable::Route r = rtable->route(q.net);
    struct sockaddr_in* server = r.second;
    std::string dest  = r.first;
    if (::connect(sockfd,(struct sockaddr *) server ,sizeof(sockaddr_in)) < 0)
    {
        utill_log(std::string(__FILE__) + " Transport: connect failed for net " + q.net + " Errno " + utill_to_string(errno));
        ::close(sockfd);
        return false;
    }

    pland::utill_io_write(sockfd, (char*)dest.c_str(), q.net.length() + 1);
    pland::utill_io_read(sockfd, responce, 3);
    responce[2] = '\0';
    std::string strResp(responce);
    if(strResp == "OK")
    {
        utill_log(std::string(__FILE__) + " Transport: Speaker connected to net " + q.net);
        Speaker *w = new Speaker(sockfd, q.queue);
        boost::unique_lock< boost::shared_mutex > lock(workerManipulator_);
        workItems_.push_back(std::make_pair(new boost::mutex(), w));
    }
    else
    {
        utill_log(std::string(__FILE__) + " Transport no queue found. Errno " + utill_to_string(errno));
        ::close(sockfd);
        return false;
    }
    return true;
}

void Transport::workerThread_()
{
    try
    {
        utill_log(std::string(__FILE__) + " Worker thread starting up.");
        while(true)
        {
            bool workedThisRound = false;
            WorkItemStat stat;
            boost::upgrade_lock< boost::shared_mutex > lock(workerManipulator_);
            for(size_t i = 0 ; i <  workItems_.size() ; i++)
            {
                boost::this_thread::interruption_point();
                if(workItems_[i].first->try_lock())
                {
                    try
                    {
                        if(workItems_[i].second->ready())
                        {
                            stat = workItems_[i].second->work();
                            statisticsLock_.lock();
                            transportStatistics_.push_back(stat);
                            statisticsLock_.unlock();
                            workedThisRound = true;
                        }
                        workItems_[i].first->unlock();
                    }
                    catch(pland::Exeption e)
                    {
                        boost::upgrade_to_unique_lock< boost::shared_mutex > uniqueLock(lock);
                        workItems_[i].first->unlock();
                        utill_log(std::string(__FILE__) + "Worker: Contact lost. Errno " + utill_to_string(errno));
                        if(workItems_[i].second->type() == WorkItem::IN)
                            delete (Listner*)workItems_[i].second;
                        else if(workItems_[i].second->type() == WorkItem::OUT)
                            delete (Speaker*)workItems_[i].second;
                        delete workItems_[i].first;
                        workItems_.erase(workItems_.begin() + i);
                        break;
                    }

                }
            }
            if(!workedThisRound)
            {
                boost::this_thread::sleep(boost::posix_time::microseconds(30));
            }
        }
    }
    catch(boost::thread_interrupted e)
    {
        utill_log(std::string(__FILE__) + " Worker thread stoping.");
        return;
    }
    return;

}

std::list <kport::WorkItemStat> Transport::getTraceBack()
{
    statisticsLock_.lock();
    std::list <WorkItemStat> ret;
    std::copy(transportStatistics_.begin(), transportStatistics_.end(), ret.begin());
    statisticsLock_.unlock();
    return ret;
}

uint Transport::getAvgLatency()
{
    statisticsLock_.lock();
    uint64_t latency = 0;
    uint packets = 0;
    for(uint i = 0; i < transportStatistics_.size(); i++)
    {
        if(transportStatistics_[i].secLatency == 0 && transportStatistics_[i].usecLatency == 0)
        {
            continue;
        }
        latency += transportStatistics_[i].secLatency*1e6;
        latency += transportStatistics_[i].usecLatency;
        packets++;
    }
    statisticsLock_.unlock();
    if(packets == 0)
        return 0;
    return latency/packets;
}

uint Transport::getAvgThroughput()
{
    statisticsLock_.lock();
    uint64_t bytes = 0;
    for(uint i = 0; i < transportStatistics_.size(); i++)
    {
        bytes += transportStatistics_[i].bytes;
    }
    statisticsLock_.unlock();
    if(transportStatistics_.size() == 0)
        return 0;
    struct timeval now;
    gettimeofday(&now, NULL);
    uint64_t microsec = now.tv_sec*1e6 + now.tv_usec;
    microsec -= transportStatistics_[0].secTime*1e6 + transportStatistics_[0].usecTime;
    if(microsec == 0)
        return 0;
    return (1e6*bytes/microsec)/2;
}
